package com.tdy.cdc.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.serializer.ValueFilter;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.nio.charset.StandardCharsets;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: NanHuang
 * @Date: 2023/04/04/11:32
 * @Description:
 */
public class MyKafkaDeserialization implements KafkaDeserializationSchema<String> {
    public static final ValueFilter FILTER = (obj, s, v) -> {
        if(v==null)
            return "";
        return v;
    };
    @Override
    public boolean isEndOfStream(String nextElement) {
        return false;
    }

    @Override
    public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
        // TODO 改进：可以直接获取每条数据对应的Topic，不需要特别的定义"table_name"
        byte[] value = record.value();
        if (value != null){
            JSONObject obj = JSONObject.parseObject(new String(value, StandardCharsets.UTF_8));
            obj.put("table_name",record.topic());
            return JSON.toJSONString(obj,FILTER, SerializerFeature.WriteMapNullValue);
        }
        return null;
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return TypeInformation.of(String.class);
    }
}
